package com.carrot.mq.mqserver;

import com.carrot.mq.common.*;
import com.carrot.mq.mqserver.core.BasicProperties;

import java.io.*;
import java.net.ServerSocket;
import java.net.Socket;
import java.net.SocketException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * @author 兴趣使然黄小黄
 * @version 1.0
 * @date 2023/8/28 20:26
 * 消息队列的本体服务器
 * 本质上是一个 TCP 服务器
 */
public class BrokerServer {

    private ServerSocket serverSocket = null;

    // 方便起见, 一个服务器只有一个虚拟主机
    private VirtualHost virtualHost = new VirtualHost("default");

    // 表示当前所有会话, 即正在与服务第进行通信的客户端
    // 此处的 key 为 channelId, value 为对应的 socket 对象
    private ConcurrentHashMap<String, Socket> sessions = new ConcurrentHashMap<>();

    // 用线程池来处理多个客户端的请求
    private ExecutorService executorService = null;

    // 控制服务器是否继续运行
    private volatile boolean runnable = true;

    public BrokerServer(int port) throws IOException {
        serverSocket = new ServerSocket(port);
    }

    /**
     * 启动 BrokerServer
     * @throws IOException
     */
    public void start() throws IOException {
        System.out.println("[BrokerServer] 启动!");
        executorService = Executors.newCachedThreadPool();
        try {
            while (runnable) {
                // 获取客户端连接
                Socket clientSocket = serverSocket.accept();
                // 处理连接的逻辑丢给线程池
                executorService.submit(() -> {
                    processConnection(clientSocket);
                });
            }
        } catch (SocketException e) {
            System.out.println("[BrokerServer] 服务器停止运行!");
        }
    }

    /**
     * 用于关闭 BrokerServer
     * 实际关闭服务只需要 kill 相应的进程
     * 这里设置是需要方便单元测试
     */
    public void stop() throws IOException {
        runnable = false;
        // 线程池中的任务放弃, 线程销毁
        executorService.shutdown();
        serverSocket.close();
    }

    /**
     * 处理一个客户端的连接
     * 可能会有多个请求和响应
     * @param clientSocket 客户端连接
     */
    private void processConnection(Socket clientSocket) {
        try(InputStream inputStream = clientSocket.getInputStream();
            OutputStream outputStream = clientSocket.getOutputStream()) {
            try (DataInputStream dataInputStream = new DataInputStream(inputStream);
                 DataOutputStream dataOutputStream = new DataOutputStream(outputStream)) {
                while (true) {
                    // 读取请求并解析
                    Request request = readRequest(dataInputStream);
                    // 根据请求计算响应
                    Response response = process(request, clientSocket);
                    // 响应写回客户端
                    writeResponse(dataOutputStream, response);
                }
            }
        } catch (EOFException | SocketException e) {
            // DataInputStream 如果读取到 EOF, 就会抛出 EOFException 异常
            // 借助异常结束循环
            System.out.println("[BrokerServer] connection 关闭! 客户端的地址: "
                    + clientSocket.getInetAddress().toString() + ":" + clientSocket.getPort());
        } catch (IOException | ClassNotFoundException | MqException e) {
            System.out.println("[BrokerServer] connection 出现异常!");
            e.printStackTrace();
        } finally {
            try {
                // 关闭 socket
                clientSocket.close();
                // 一个 TCP 连接中, 可能包含多个 channel, 需要把当前 socket 对应的所有 channel 清理
                clearClosedSession(clientSocket);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 清理过期的会话
     * @param clientSocket 失效的 socket
     */
    private void clearClosedSession(Socket clientSocket) {
        // 遍历 sessions hash 表, 把该被关闭的 socket 对应的键值对删掉
        List<String> toDeleteChannelId = new ArrayList<>(); // 用于记录要删除的 channelId
        for (Map.Entry<String, Socket> entry : sessions.entrySet()) {
            if (entry.getValue() == clientSocket) {
                // 防止迭代器失效, 不能一边遍历, 一边删除, 这里使用记录的方式
                toDeleteChannelId.add(entry.getKey());
            }
        }
        // 删除
        for (String channelId : toDeleteChannelId) {
            sessions.remove(channelId);
        }
        System.out.println("[BrokerServer] 清理 session 完成! 被清理的 channelId=" + toDeleteChannelId);
    }

    /**
     * 响应写回客户端
     * @param dataOutputStream 写入的位置
     * @param response 响应
     */
    private void writeResponse(DataOutputStream dataOutputStream, Response response) throws IOException {
        dataOutputStream.writeInt(response.getType());
        dataOutputStream.writeInt(response.getLength());
        dataOutputStream.write(response.getPayload());
        // 刷新缓冲区
        dataOutputStream.flush();
    }

    /**
     * 处理请求并返回响应
     * @param request 请求
     * @param clientSocket clientSocket
     * @return 处理后得到的响应
     */
    private Response process(Request request, Socket clientSocket) throws IOException, ClassNotFoundException, MqException {
        // 对 request 中的 payload 进行解析
        BasicArguments basicArguments = (BasicArguments) BinaryTool.fromBytes(request.getPayload());
        System.out.println("[Request] rid=" + basicArguments.getRid() + ", channelId=" + basicArguments.getChannelId()
            + ", type=" + request.getType() + ", length=" + request.getLength());
        // 根据 type 的值区分请求的目的
        boolean ok = true;
        switch (request.getType()) {
            case 0x1: // 创建 channel
                sessions.put(basicArguments.getChannelId(), clientSocket);
                System.out.println("[BrokerServer] 创建 channel 完成! channelId=" + basicArguments.getChannelId());
                break;
            case 0x2: // 销毁 channel
                sessions.remove(basicArguments.getChannelId());
                System.out.println("[BrokerServer] 销毁 channel 完成! channelId=" + basicArguments.getChannelId());
                break;
            case 0x3: // 创建交换机
                ExchangeDeclareArguments exchangeDeclareArguments = (ExchangeDeclareArguments) basicArguments;
                ok = virtualHost.exchangeDeclare(exchangeDeclareArguments.getExchangeName(),
                        exchangeDeclareArguments.getExchangeType(),
                        exchangeDeclareArguments.isDurable(),
                        exchangeDeclareArguments.isAutoDelete(), exchangeDeclareArguments.getArguments());
                break;
            case 0x4: // 销毁交换机
                ExchangeDeleteArguments exchangeDeleteArguments = (ExchangeDeleteArguments) basicArguments;
                ok = virtualHost.exchangeDelete(exchangeDeleteArguments.getExchangeName());
                break;
            case 0x5: // 创建队列
                QueueDeclareArguments queueDeclareArguments = (QueueDeclareArguments) basicArguments;
                ok = virtualHost.queueDeclare(queueDeclareArguments.getQueueName(),
                        queueDeclareArguments.isDurable(), queueDeclareArguments.isExclusive(),
                        queueDeclareArguments.isAutoDelete(), queueDeclareArguments.getArguments());
                break;
            case 0x6: // 销毁队列
                QueueDeleteArguments queueDeleteArguments = (QueueDeleteArguments) basicArguments;
                ok = virtualHost.queueDelete(queueDeleteArguments.getQueueName());
                break;
            case 0x7: // 创建 binding
                QueueBindArguments queueBindArguments = (QueueBindArguments) basicArguments;
                ok = virtualHost.queueBind(queueBindArguments.getQueueName(),
                        queueBindArguments.getExchangeName(), queueBindArguments.getBindingKey());
                break;
            case 0x8: // 删除 binding
                QueueUnbindArguments queueUnbindArguments = (QueueUnbindArguments) basicArguments;
                ok = virtualHost.queueUnbind(queueUnbindArguments.getQueueName(), queueUnbindArguments.getExchangeName());
                break;
            case 0x9: // 发送消息
                BasicPublishArguments basicPublishArguments = (BasicPublishArguments) basicArguments;
                ok = virtualHost.basicPublish(basicPublishArguments.getExchangeName(),
                        basicPublishArguments.getRoutingKey(), basicPublishArguments.getBasicProperties(),
                        basicPublishArguments.getBody());
                break;
            case 0xa: // 订阅消息
                BasicConsumeArguments basicConsumeArguments = (BasicConsumeArguments) basicArguments;
                ok = virtualHost.basicConsume(basicConsumeArguments.getConsumerTag(),
                        basicConsumeArguments.getQueueName(), basicConsumeArguments.isAutoAck(),
                        new Consumer() {
                            // 该回调方法的作用, 就是把服务器收到的消息直接推送回对应的消费者客户端
                            @Override
                            public void handleDelivery(String consumerTag, BasicProperties basicProperties, byte[] body) throws MqException, IOException {
                                // 确认当前收到的消息需要发给的客户端
                                // 此处的 consumerTag 实际上是 channelId, 根据其可以查询对应的 sessions 获得 socket 对象
                                Socket clientSocket = sessions.get(consumerTag);
                                if (clientSocket == null || clientSocket.isClosed()) {
                                    throw new MqException("[BrokerServer] 订阅消息的客户端已经关闭!");
                                }
                                // 构造响应数据
                                SubScribeReturns subScribeReturns = new SubScribeReturns();
                                subScribeReturns.setChannelId(consumerTag);
                                subScribeReturns.setRid(""); // 只有响应没有请求, rid 暂时不需要
                                subScribeReturns.setOk(true);
                                subScribeReturns.setConsumerTag(consumerTag);
                                subScribeReturns.setBasicProperties(basicProperties);
                                subScribeReturns.setBody(body);
                                byte[] payload = BinaryTool.toBytes(subScribeReturns);
                                Response response = new Response();
                                // 0xc 表示服务器给消费者客户端推送的消息数据
                                response.setType(0xc);
                                response.setLength(payload.length);
                                response.setPayload(payload);
                                // 写入响应, 把数据写回客户端
                                DataOutputStream dataOutputStream = new DataOutputStream(clientSocket.getOutputStream());
                                writeResponse(dataOutputStream, response);
                            }
                        });
                break;
            case 0xb: // 确认消息
                BasicAckArguments basicAckArguments = new BasicAckArguments();
                ok = virtualHost.basicAck(basicAckArguments.getQueueName(), basicAckArguments.getMessageId());
                break;
            default: // type 非法
                throw new MqException("[BrokerServer] 未知的 type! type=" + request.getType());
        }
        // 构造响应
        BasicReturns basicReturns = new BasicReturns();
        basicReturns.setChannelId(basicArguments.getChannelId());
        basicReturns.setRid(basicArguments.getRid());
        basicReturns.setOk(ok);
        byte[] payload = BinaryTool.toBytes(basicReturns);
        Response response = new Response();
        response.setType(request.getType());
        response.setLength(payload.length);
        response.setPayload(payload);
        System.out.println("[Response] rid=" + basicArguments.getRid() + ", channelId=" + basicArguments.getChannelId()
            + ", type=" + response.getType() + ", length=" + response.getLength());
        return response;
    }

    /**
     * 读取请求对象
     * @return 返回读取的请求对象
     * @throws IOException 读取请求时, payload 长度与实际不符
     */
    private Request readRequest(DataInputStream dataInputStream) throws IOException {
        Request request = new Request();
        request.setType(dataInputStream.readInt());
        request.setLength(dataInputStream.readInt());
        byte[] payload = new byte[request.getLength()];
        int n = dataInputStream.read(payload);
        if (n != request.getLength()) {
            throw new IOException("读取请求格式出错!");
        }
        request.setPayload(payload);
        return request;
    }
}
